package com.linys.scala.qf.day09_DAG_CACHE

import java.net.URL

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/*
用户点击产生日志信息,有时间戳和对应URL
URL中会有不用学科,统计学科的访问量
需求:根据用访问数据进行统计用户对各个学科的各个模块的访问量Top3
思路:1.统计每个模块的访问量
2. 按照学科进行分组
3. 学科进行排序
4. 取top3
*/
object SubjectAccess {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getName).setMaster("local[2]")
    val sc = new SparkContext(conf)

    // 获取数据
    val logs = sc.textFile("D://IDEA/ScalaDemo/src/main/scala/com/linys/scala/qf/day09_DAG_CACHE/data/access.txt")

    // 切分并生成元组，便于聚合
    val tups: RDD[(String, Int)] = logs.map(line => {
      val url = line.split("\t")(1)
      (url, 1)
    })

    println(tups.collect.toBuffer)
    println("-------------------------")

    // 将相同的url进行聚合，求出各个模块的访问量
    val aggred: RDD[(String, Int)] = tups.reduceByKey(_+_)
    println(aggred.collect.toBuffer)
    println("-------------------------")

    // 接下来需要拿到每个学科，然后按照学科进行分组
    val subjectAndUrlAndCount: RDD[(String, (String, Int))] = aggred.map(tup => {
      val url = tup._1 // 用户请求的url
      val count = tup._2 // 模块的访问量
      val subject = new URL(url).getHost // 学科信息
      (subject, (url, count))
    })
    println(subjectAndUrlAndCount.collect.toBuffer)
    println("-------------------------")

    // 按照学科进行分组
    val grouped: RDD[(String, Iterable[(String, Int)])] = subjectAndUrlAndCount.groupByKey
    grouped.cache
    println(grouped.collect.toBuffer)
    println("-------------------------")

    // 组内排序并取top1
    val res: RDD[(String, List[(String, Int)])] = grouped.mapValues(_.toList.sortWith(_._2 > _._2).take(1))
    println(res.collect.toBuffer)
    val res2: RDD[(String, List[(String, Int)])] = grouped.mapValues(_.toList.sortWith(_._2 > _._2))
    println(res2.collect.toBuffer)

    sc.stop

  }
}
